package com.idanchuang.component.redis.listener;

import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.models.partitions.RedisClusterNode;
import io.lettuce.core.cluster.pubsub.RedisClusterPubSubAdapter;
import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection;
import io.lettuce.core.cluster.pubsub.api.sync.RedisClusterPubSubCommands;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutorService;

public class RedisClusterKeyEventExpiredListener
        extends RedisClusterPubSubAdapter<String, String> {

    private static final Logger logger = LoggerFactory.getLogger(RedisClusterKeyEventExpiredListener.class);

    private static final String KEYSPACE_PATTERN = "__keyevent*__:expired";

    private final RedisClusterClient redisClusterClient;
    private final KeyExpiredProcessor processor;

    public RedisClusterKeyEventExpiredListener(RedisClusterClient redisClusterClient, KeyExpiredProcessor processor) {
        this.redisClusterClient = redisClusterClient;
        this.processor = processor;
    }

    @PostConstruct
    public void init() {
        StatefulRedisClusterPubSubConnection<String, String> connection = redisClusterClient.connectPubSub();
        connection.addListener(this);
        connection.setNodeMessagePropagation(true);
        RedisClusterPubSubCommands<String, String> sync = connection.sync();
        sync.masters().commands().psubscribe(KEYSPACE_PATTERN);
    }

    @Override
    public void message(RedisClusterNode node, String pattern, String channel,
                        String message) {
        ExecutorService worker = processor.procMode() == ProcMode.SERIAL ?
                RedisKeyEventExpiredListener.SERIAL_WORKER : RedisKeyEventExpiredListener.CONCURRENT_WORKER;
        worker.execute(() -> {
            try {
                processor.process(message.getBytes());
            } catch (Exception e) {
                logger.error("doHandleMessage error", e);
            }
        });
    }
}
